package com.sbwebsocket.core.redis;

import cn.hutool.core.collection.CollUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sbwebsocket.core.model.bo.TopicMessageBody;
import com.sbwebsocket.core.model.bo.UnusualMessage;
import com.sbwebsocket.core.redis.model.contants.WsMessageRedisKeyPrefix;
import com.sbwebsocket.core.ws.WebSocketServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Auther: GSJ
 * @Date: 2022/2/16
 * @Description:
 * @version: 1.0
 */
@Slf4j
public class RedisTopicMessageReceiver {

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private RedisDelayProcUnusualMessage redisDelayProcUnusualMessage;

    @Autowired
    private StringRedisTemplate redisTemplate;

    public void receive(String jsonMsg){
        log.info("开始接收fanout消息:{}",jsonMsg);
        /**
         * step1:遍历topics,查找redis,获取到对应主题下的clientKeys
         * step2:clientKeys遍历发送消息
         */
        try {
            TopicMessageBody messageBody = objectMapper.readValue(jsonMsg, TopicMessageBody.class);
            ConcurrentHashMap<String, WebSocketServer> webSocketMap = WebSocketServer.webSocketMap;
            for (String topic : messageBody.getTopic()) {
                Set<String> clientKeys = redisTemplate.opsForSet().members(WsMessageRedisKeyPrefix.SERVER_MSG_TOPIC + topic);
                if(CollUtil.isEmpty(clientKeys)) return;
                for (String clientKey : clientKeys) {
                    WebSocketServer webSocketServer = webSocketMap.get(clientKey);
                    if(null!=webSocketServer){
                        if(log.isDebugEnabled()){
                            log.debug("给客户端clientKey:{}发送消息完成！",clientKey);
                        }
                        try {
                            webSocketServer.sendMessage(messageBody.getBody());
                        } catch (Exception e) {
                            log.debug("client异常离线clientKey:{}",clientKey);
                            redisDelayProcUnusualMessage.save(new UnusualMessage(clientKey,messageBody.getBody()));
                        }
                    }
                }
                
            }
                
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

    }
}
